package org.example;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.http.client.methods.HttpPost;
import sun.net.www.http.HttpClient;

import java.io.File;
import java.io.FileInputStream;
import java.io.FileOutputStream;
import java.io.IOException;
import java.net.URI;
import java.net.URISyntaxException;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;

/**
 * @author zhl
 * @version 1.0
 * @ClassName
 * @date 2020/11/2 22:09
 */
public class LogKeeper {

    private String path;         // Tomcat log 地址
    private String HDFSUri;     // hadoop地址
    private String tomcatBaseDir = "/user/hadoop/tomcatLog";                // hadoop中保存log的文件地址
    private String HADOOP_USER_NAME = "hadoop";
    private Map<String, Long> hadoopFileModifierTime = new ConcurrentHashMap<>();           // 保存hadoop文件修改时间
    private int delay = 0;          // log保存延迟时间
    private int period = 5000;      // log保存间隔时间(毫秒)
    private Configuration conf = new Configuration();

    public LogKeeper() {
        this("D:\\Tomcat8\\Tomcat8\\logs", "hdfs://192.168.136.132:9000");
    }

    public LogKeeper(String originPath) {
        this(originPath, "hdfs://192.168.136.132:9000");
    }

    /**
     *
     * @param originPath 原地址，在项目中为tomcat日志地址
     * @param dstPath   目标地址，在项目中为hadoop地址
     * */
    public LogKeeper(String originPath, String dstPath) {
        this.path = originPath;
        this.HDFSUri = dstPath;
        System.setProperty("HADOOP_USER_NAME", "hadoop");
        conf.set("fs.defaultFS", "hdfs://192.168.136.132:9000");
    }

    public String getHADOOP_USER_NAME() {
        return HADOOP_USER_NAME;
    }

    public void setHADOOP_USER_NAME(String HADOOP_USER_NAME) {
        this.HADOOP_USER_NAME = HADOOP_USER_NAME;
    }

    public Map<String, Long> getHadoopFileModifierTime() {
        return hadoopFileModifierTime;
    }

    public void setHadoopFileModifierTime(Map<String, Long> hadoopFileModifierTime) {
        this.hadoopFileModifierTime = hadoopFileModifierTime;
    }

    public String getTomcatBaseDir() {
        return tomcatBaseDir;
    }

    public void setTomcatBaseDir(String tomcatBaseDir) {
        this.tomcatBaseDir = tomcatBaseDir;
    }

    public String getPath() {
        return path;
    }

    public String getHDFSUri() {
        return HDFSUri;
    }

    public void setHDFSUri(String HDFSUri) {
        this.HDFSUri = HDFSUri;
    }

    public void setPath(String path) {
        this.path = path;
    }

    public int getDelay() {
        return delay;
    }

    public void setDelay(int delay) {
        this.delay = delay;
    }

    public int getPeriod() {
        return period;
    }

    public void setPeriod(int period) {
        this.period = period;
    }

    // 判断文件是否需要再次保存到hadoop中
    private boolean needSave(File file) {

        // 测试，通过0获取Date
//        SimpleDateFormat sdf= new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
//        java.util.Date date = new Date(hadoopFileModifierTime.get(file.getName()));
//        String str = sdf.format(date);
//        System.out.println(str);

        // 如果文件不存在，则需要保存
        if (hadoopFileModifierTime.get(file.getName()) == null) {
            return true;
        }
        // 判断最后修改时间是否一致，一致则不需要保存，不一致则进行保存
        return hadoopFileModifierTime.get(file.getName()) < file.lastModified();
    }

    // 从path中分理出文件名称
    private String getFileName(String path) {
        int index = path.lastIndexOf("/");
        if (index == -1) {
            return "";
        } else {
            return path.substring(index+1);
        }
    }

    // 更新本地hadoop文件修改时间缓存
    private void refreshHadoopFileList() throws IOException{
        Path path = new Path(HDFSUri+tomcatBaseDir);
        FileSystem fs = FileSystem.get(URI.create(HDFSUri+"/"+tomcatBaseDir), conf);
        FileStatus[] hadoopFileStatus = fs.listStatus(path);
        for (FileStatus fileStatus : hadoopFileStatus) {
            // key 为文件名称，value为最后修改时间
            hadoopFileModifierTime.putIfAbsent(getFileName(fileStatus.getPath().toString()),
                    fileStatus.getModificationTime());
        }
    }

    // 读取file内容保存到FileSystem
    private void save(File file){
        FileSystem fs = null;
        FSDataOutputStream fsdOS = null;
        byte[] buffer = new byte[1024];
        int len = 0;
        FileInputStream fis = null;
        try {
            fs = FileSystem.get(new URI(HDFSUri+tomcatBaseDir+"/"+file.getName()), conf, HADOOP_USER_NAME);
            fsdOS = fs.create(new Path(HDFSUri+tomcatBaseDir+"/"+file.getName()));
            fis = new FileInputStream(file);
            while ((len = fis.read(buffer)) != -1) {
                fsdOS.write(buffer, 0, len);
            }
        }catch (IOException | URISyntaxException | InterruptedException e){
            e.printStackTrace();
        }
    }

    // 保存tomcat的log到hadoop
    private void keepLog() {
        System.out.println("保存开始");
        try {
            refreshHadoopFileList();
        }catch (Exception e) {
            e.printStackTrace();
        }
        File file = new File(path);
        File[] files = file.listFiles();    // 获取tomcat目录下所有log文件
        // 没有log就不进行保存
        if (files == null) {
            return;
        }
        int count = 0;
        // 判断文件是否需要进行保存，如果需要启动多线程保存
        for (final File temp : files) {
            if (needSave(temp)) {
                count++;
                new Thread(() -> save(temp)).start();
            }
        }
        System.out.println(count+"文件修改");
        System.out.println("保存结束");
    }

    public static class LogBuilder extends LogKeeper{
        private LogKeeper logKeeper = new LogKeeper();

        public LogBuilder path(String path) {
            logKeeper.setPath(path);
            return this;
        }

        public LogBuilder HDFSUri(String HDFSUri) {
            logKeeper.setHDFSUri(HDFSUri);
            return this;
        }

        public LogBuilder tomcatBaseDir(String tomcatBaseDir) {
            logKeeper.setTomcatBaseDir(tomcatBaseDir);
            return this;
        }

        public LogBuilder HADOOP_USER_NAME(String HADOOP_USER_NAME) {
            logKeeper.setHADOOP_USER_NAME(HADOOP_USER_NAME);
            return this;
        }

        public LogBuilder delay(int delay) {
            logKeeper.setDelay(delay);
            return this;
        }

        public LogBuilder period(int period) {
            logKeeper.setPeriod(period);
            return this;
        }
    }

    // tomcat的log保存程序运行
    public void run() {
        new Timer().schedule(new TimerTask() {
            @Override
            public void run() {
                keepLog();
            }
        }, delay, period);
    }

    public static void main(String[] args) {
        LogKeeper logKeeper = new LogBuilder();
        logKeeper.run();
    }
}
